#include "index/attribute_index/attribute_mem_indexer/dump/var_len_data_dumper.h"

namespace hawking {
namespace indexlib {

// record the offset in varlen_metafile for every doc
Status VarLenDataDumper::SimpleOffsetFileWrite_(size_t doc_count) {
    std::vector<size_t> offsets;
    offsets.resize(doc_count, 0);

    size_t start_offset = 0;
    size_t length = sizeof(size_t) + sizeof(ArrayLengthType);
    for (size_t idx = 0; idx < doc_count; ++idx) {
        offsets[idx] = start_offset + idx * length;
    }

    Status ret = offset_file_writer_->Write(
        (uint8_t*)(offsets.data()), offsets.length() * sizeof(size_t), 0).Code();
    offset_file_writer_.Close();
    return ret;
}

Status VarLenDataDumper::DumpMultiT_(
    const std::unordered_map<DocId, DocId>& old2new_map, uint32_t element_size) {
    // write offset file
    if (SimpleOffsetFileWrite_(old2new_map.size()) != Status::OK) {
        LOG(ERROR) << "dump offset file fail";
        return ret;
    }

    // doc_count * (offset_in_datafile(8) + elem_count(2 in default))
    size_t varlen_metafile_length =
        old2new_map.size() * (sizeof(size_t) + sizeof(ArrayLengthType));
    varlen_metafile_writer_->ReserveFile(varlen_metafile_length);
    VarlenMetaBuffer varlen_metafile_buffer_writer(varlen_metafile_length);

    // write varlen_metafile and calc datafile length
    size_t datafile_length = 0;
    std::shared_ptr<VarLenDataIterator> iter = accessor_->CreateDataIterator();
    while (iter->HasNext()) {
        uint8_t* data = nullptr;
        iter->Next();
        if (old2new_map.find(iter->Cursor()) == old2new_map.end()) {
            continue;
        }

        iter->CurrentData(std::ref(data));
        ArrayLengthType elem_count = *((ArrayLengthType*)data);
        // offset in datafile
        varlen_metafile_buffer_writer->AppendOffset(datafile_length);
        // elem_count
        varlen_metafile_buffer_writer->AppendElementCount(elem_count);
        datafile_length += (element_size * elem_count);
    }

    // write varlen_metafile with one syscall
    auto ret = varlen_metafile_writer_->Write(
        varlen_metafile_buffer_writer.Buffer(), varlen_metafile_length, 0);
    if (!ret.OK()) {
        LOG(ERROR) << "write varlen_metafile fail";
        return ret.Code();
    }
    varlen_metafile_writer_.Close();
    data_file_writer_->ReserveFile(datafile_length);

    // write data file
    iter->Reset();
    datafile_length = 0;
    while (iter->HasNext()) {
        iter->Next();
        if (old2new_map.find(iter->Cursor()) == old2new_map.end()) {
            continue;
        }

        uint8_t* data = nullptr;
        iter->CurrentData(std::ref(data));
        ArrayLengthType elem_count = *((ArrayLengthType*)data);
        data += sizeof(ArrayLengthType);
        ret = data_file_writer_->Write(data, (elem_count * element_size), datafile_length);
        if (!ret.OK()) {
            LOG(ERROR) << "write datafile fail";
            return ret.Code();
        }
        datafile_length += (element_size * elem_count);
    }

    data_file_writer_->Close();
    return Status::OK;
}

Status VarLenDataDumper::DumpMultiString_(
    const std::unordered_map<DocId, DocId>& old2new_map) {
    // calc the offsets and the length of varlen_metafile and datafile
    std::vector<size_t> offsets;
    offsets.reserve(old2new_map.size());
    size_t varlen_metafile_length = 0;
    size_t datafile_length = 0;
    std::shared_ptr<VarLenDataIterator> iter = accessor_->CreateDataIterator();
    while (iter->HasNext()) {
        iter->Next();
        if (old2new_map.find(iter->Cursor()) == old2new_map.end()) {
            continue;
        }

        uint8_t* data = nullptr;
        iter->CurrentData(std::ref(data));
        ArrayLengthType elem_count = *((ArrayLengthType*)data);
        data += sizeof(elem_count);

        offsets.emplace_back(varlen_metafile_length);
        varlen_metafile_length += sizeof(size_t);
        varlen_metafile_length += sizeof(elem_count);
        varlen_metafile_length += (elem_count * sizeof(StringLengthType));
        for (ArrayLengthType idx = 0; idx < elem_count; ++idx) {
            StringLengthType str_length = *((StringLengthType*)data);
            data += sizeof(str_length);
            data += str_length;
            datafile_length += str_length;
        }
    }

    // write offset file
    Status ret = offset_file_writer_->Write(
        (uint8_t*)(offsets.data()), offsets.length() * sizeof(size_t), 0).Code();
    if (ret != Status::OK) {
        LOG(ERROR) << "dump offset file fail";
        return ret;
    }

    // reserve varlen_metafile length and datafile
    data_file_writer_->ReserveFile(datafile_length);
    varlen_metafile_writer_->ReserveFile(varlen_metafile_length);
    VarlenMetaBuffer varlen_metafile_buffer_writer(varlen_metafile_length);

    // write datafile and append varlen_metafile's content
    datafile_length = 0;
    iter.Reset();
    while (iter->HasNext()) {
        iter->Next();
        if (old2new_map.find(iter->Cursor()) == old2new_map.end()) {
            continue;
        }

        uint8_t* data = nullptr;
        iter->CurrentData(std::ref(data));
        ArrayLengthType elem_count = *((ArrayLengthType*)data);
        data += sizeof(elem_count);

        varlen_metafile_buffer_writer->AppendOffset(datafile_length);
        varlen_metafile_buffer_writer->AppendElementCount(elem_count);
        for (ArrayLengthType idx = 0; idx < elem_count; ++idx) {
            // append every string length
            StringLengthType str_length = *((StringLengthType*)data);
            varlen_metafile_buffer_writer->AppendStringLength(str_length);
            data += sizeof(str_length);

            // write every string to datafile
            ret = data_file_writer_->Write(data, str_length, datafile_length);
            if (ret != Status::OK) {
                LOG(ERROR) << "write datafile fail";
                return ret.Code();
            }
            data += str_length;
            datafile_length += str_length;
        }
    }

    // write varlen_metafile
    ret = varlen_metafile_writer_->Write(
        varlen_metafile_buffer_writer.Buffer(), varlen_metafile_length, 0);
    if (!ret.OK()) {
        LOG(ERROR) << "write varlen_metafile fail";
        return ret.Code();
    }
}

/*
    multiT:
    offsetfile: |  doc1's offset in varlen_metafile | doc2's offset in varlen_metafile | .....
    varlen_metafile: | doc1's offset in datafile + array-length | doc2's offset in datafile + array-length | .....
    datafile: | doc1's every element | doc2's every element | .....

    multiString:
    offsetfile: |  doc1's offset in varlen_metafile | doc2's offset in varlen_metafile | .....
    varlen_metafile: | doc1's offset in datafile + element_length + every string's length | .....
    datafile: | doc1's every string | doc2's every string | .....

    Q: why need varlen_metafile?
    A: when online update one doc's attribute, only update offsetfile with one `size_t`(thread safe)
        to notify it is already not in datafile.
*/
Status VarLenDataDumper::Dump(
    util::ColumnType field_type,
    uint32_t element_size,
    const std::string& offset_file,
    const std::string& varlen_metafile,
    const std::string& data_file,
    const std::unordered_map<DocId, DocId>& old2new_map) {
    if (!accessor_) {
        data_file_writer_->Close();
        offset_file_writer_->Close();
        return Status::InternalError;
    }

    data_file_writer_ = std::make_unique<NormalFileWriter>(data_file);
    varlen_metafile_writer_ = std::make_unique<NormalFileWriter>(varlen_metafile);
    offset_file_writer_ = std::make_unique<NormalFileWriter>(offset_file);
    Status ret = data_file_writer_->Init();
    if (ret != Status::OK) {
        LOG(ERROR) << "attribute data file " << data_file << " Init fail";
        return ret;
    }
    ret = offset_file_writer_->Init();
    if (ret != Status::OK) {
        LOG(ERROR) << "attribute offset file " << offset_file << " Init fail";
        return ret;
    }
    ret = varlen_metafile_writer_->Init();
    if (ret != Status::OK) {
        LOG(ERROR) << "attribute varlen metafile " << offset_file << " Init fail";
        return ret;
    }

    offset_file_writer_->ReserveFile(sizeof(size_t) * old2new_map.size());
    return (field_type == util::ColumnType::COLUMN_STRING_LIST)
        ? DumpMultiString_() : DumpMultiT_(old2new_map, element_size);
}

}
}